package controller

import (
	"context"

	"gitee.com/zatto/maudit/apps/audit"
	"gitee.com/zatto/maudit/common/logger"
	"gitee.com/zatto/maudit/conf"
	"github.com/infraboard/mcube/app"
	"github.com/segmentio/kafka-go"
)

func NewAuditLogSaveController(kc *conf.Kafka) *AuditLogSaveController {
	return &AuditLogSaveController{
		kc:    kc,
		audit: app.GetInternalApp(audit.AppName).(audit.Service),
	}
}

type AuditLogSaveController struct {
	kc    *conf.Kafka
	r     *kafka.Reader
	audit audit.Service
}

func (c *AuditLogSaveController) Run(ctx context.Context) {
	c.r = kafka.NewReader(kafka.ReaderConfig{
		Brokers:  c.kc.Endpoints,
		GroupID:  c.kc.ConsumerGroup,
		Topic:    c.kc.Topic,
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
	})

	for {
		m, err := c.r.ReadMessage(ctx)
		if err != nil {
			break
		}
		logger.L().Debug().Msgf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
		ins, err := audit.LoadAuditLogFromJson(m.Value)
		if err != nil {
			logger.L().Error().Msgf("load audit log error, %s", err)
			continue
		}
		_, err = c.audit.SaveAuditLog(ctx, ins)
		if err != nil {
			logger.L().Error().Msgf("save audit log error, %s", err)
			continue
		}
	}
}

func (c *AuditLogSaveController) Stop() error {
	return c.r.Close()
}
